package cn.getech.data.development.service.impl;

import cn.getech.data.development.constant.LastRunState;
import cn.getech.data.development.dto.JobInfoDto;
import cn.getech.data.development.dto.JobLinkAndJobDto;
import cn.getech.data.development.dto.JobLinkDto;
import cn.getech.data.development.entity.*;
import cn.getech.data.development.mapper.*;
import cn.getech.data.development.service.WorkFlowDepService;
import cn.getech.data.development.utils.DataPermissionUtil;
import cn.getech.system.center.utils.ShiroUtils;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DateUtil;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;

@Slf4j
@Service
public class WorkFlowDepServiceImpl extends ServiceImpl<WorkFlowDepMapper, WorkFlowDep> implements WorkFlowDepService {
    @Autowired
    private WorkFlowDepMapper workFlowDepMapper;
    @Autowired
    private JobDepRunHistoryMapper jobDepRunHistoryMapper;
    @Autowired
    private JobRunHistoryMapper jobRunHistoryMapper;
    @Autowired
    private JobNodeRunHistoryMapper jobNodeRunHistoryMapper;
    @Resource
    private ProcInfoMapper procInfoMapper;

    @Override
    public List<Integer> selectJobIds() {
        Long userId = ShiroUtils.getUserId();
        Integer tenantId = ShiroUtils.getTenantId();
        if (DataPermissionUtil.isAdmin(userId.intValue())) {
            userId = null;
        }
        List<ProcInfo> authorisedProds = procInfoMapper.listUserProcs(userId, tenantId);
        List<Integer> authorisedProdIds = authorisedProds.stream().map(ProcInfo::getId).collect(Collectors.toList());
        QueryWrapper<WorkFlowDep> query = new QueryWrapper<>();
        authorisedProdIds.add(-1);
        query.in("proc_id", new ArrayList<>(authorisedProdIds));

        List<WorkFlowDep> workFlowDeps = workFlowDepMapper.selectList(query);
        List<Integer> jobInfoIds = new ArrayList<>();
        if (CollectionUtil.isNotEmpty(workFlowDeps)) {
            workFlowDeps.forEach(workFlowDep -> {
                jobInfoIds.add(workFlowDep.getId().intValue());
            });
        }
        return jobInfoIds;
    }

    @Override
    public List<Integer> selectJobIdsByProcId(Integer procId) {
        List<WorkFlowDep> workFlowDeps = workFlowDepMapper.selectList(new QueryWrapper<WorkFlowDep>().eq("proc_id",procId));
        List<Integer> jobInfoIds = new ArrayList<>();
        if (CollectionUtil.isNotEmpty(workFlowDeps)) {
            workFlowDeps.forEach(workFlowDep -> {
                jobInfoIds.add(workFlowDep.getId().intValue());
            });
        }
        return jobInfoIds;
    }

    @Override
    public void initWorkFlowDepHistory(List<JobLinkDto> jobLinks, List<JobInfoDto> jobInfos, String confParams, Integer workFlowDepId, String oozieJobId, int runType, String queueName) {
        //初始化对应的数据
        /**
         * 第一次执行默认版本号为1，否则为 当前任务的运行日志中最最大的版本号
         */
        JobDepHistory jobDepHistory = new JobDepHistory();
        jobDepHistory.setNum(null);
        jobDepHistory.setParams(confParams);
        jobDepHistory.setState(LastRunState.RUNNING.getCode());
        jobDepHistory.setBeginTime(DateUtil.parseDateTime(DateUtil.now()));
        jobDepHistory.setOozieJobId(oozieJobId);
        jobDepHistory.setJobDepId(Long.valueOf(workFlowDepId));
        JobLinkAndJobDto jobLinkAndJobDto = new JobLinkAndJobDto();
        jobLinkAndJobDto.setEdges(jobLinks);
        jobLinkAndJobDto.setNodes(jobInfos);
        jobDepHistory.setNodeParams(JSONObject.toJSONString(jobLinkAndJobDto));
        jobDepHistory.setRunType(runType);
        jobDepHistory.setQueueName(queueName);
        jobDepRunHistoryMapper.insert(jobDepHistory);

    }

    @Override
    public void stopAllJobAndJobNodeHistoryByWorkFlowLay(Long workDlowDepId, String lastOOziedId, String lastCorOozieId) {

        WorkFlowDep workFlowDep = workFlowDepMapper.selectById(workDlowDepId);
        if (null == workFlowDep) {
            return;
        }
        if (Objects.equals(LastRunState.RUNNING.getCode(), workFlowDep.getLastRunState())) {
            workFlowDep.setLastRunState(LastRunState.FAIL.getCode());
            workFlowDepMapper.updateById(workFlowDep);
            if(StringUtils.isNotEmpty(lastOOziedId)){
                workFlowDepMapper.updateLastOozied(workDlowDepId);
            }
            if(StringUtils.isNotEmpty(lastCorOozieId)){
                workFlowDepMapper.updateLastCorOozied(workDlowDepId);
            }
        }

        //停止手动任务
        if(StringUtils.isNotEmpty(lastOOziedId)){
            //改变其历史中的记录
            List<JobDepHistory> jobDepHistories = jobDepRunHistoryMapper.selectList(new QueryWrapper<JobDepHistory>()
                    .eq("job_dep_id", workDlowDepId)
                    .eq("state", LastRunState.RUNNING.getCode())
                    .eq("oozie_job_id", lastOOziedId));
            if(CollectionUtil.isNotEmpty(jobDepHistories)){
                jobDepRunHistoryMapper.batchUpdateStatus(jobDepHistories);
            }
            //其对应的工作流节点和节点的运行状态
            List<JobRunHistory> jobRunHistories = jobRunHistoryMapper.selectList(new QueryWrapper<JobRunHistory>()
                    .eq("state", LastRunState.RUNNING.getCode())
                    .eq("dep_oozie_job_id", lastOOziedId));
            if(CollectionUtil.isNotEmpty(jobRunHistories)){
                jobRunHistoryMapper.batchUpdateKillStatus(jobRunHistories);
                for (JobRunHistory jobRunHistory : jobRunHistories) {
                    List<JobNodeRunHistory> jobNodeRunHistories = jobNodeRunHistoryMapper.selectList(new QueryWrapper<JobNodeRunHistory>()
                            .eq("job_info_id", jobRunHistory.getJobInfoId())
                            .eq("state",LastRunState.RUNNING.getCode())
                            .eq("oozie_job_id", jobRunHistory.getOozieJobId()));
                    if(CollectionUtil.isNotEmpty(jobNodeRunHistories)){
                        for (JobNodeRunHistory jobNodeRunHistory : jobNodeRunHistories) {
                            //改变其log中的运行状态
                            if(null != jobNodeRunHistory){
                                String logs = jobNodeRunHistory.getLog();
                                if(StringUtils.isNotEmpty(logs)){
                                    try {
                                        JSONObject obj = JSONObject.parseObject(logs);
                                        if(null != obj){
                                            String status = obj.getString("status");
                                            if(StringUtils.isNotEmpty(status) && status.trim().toLowerCase().equals("running")){
                                                obj.put("status","killed");
                                                jobNodeRunHistory.setLog(obj.toJSONString());
                                            }
                                        }
                                    }catch (Exception e){
                                        log.error("转换日志异常！error:{}",e.getMessage());
                                    }
                                }
                            }
                        }
                        jobNodeRunHistoryMapper.batchUpdateKillStatus(jobNodeRunHistories);
                    }
                }
            }
        }

        //停止定时任务
        if(StringUtils.isNotEmpty(lastCorOozieId)){
            //改变其历史中的记录
            List<JobDepHistory> jobDepHistories = jobDepRunHistoryMapper.selectByLimit20(workDlowDepId,10);
            if(CollectionUtil.isNotEmpty(jobDepHistories)){
                jobDepHistories = jobDepHistories.stream().filter(o -> Objects.equals(o.getState(),LastRunState.RUNNING.getCode())).collect(Collectors.toList());
            }
            if(CollectionUtil.isNotEmpty(jobDepHistories)){
                jobDepRunHistoryMapper.batchUpdateStatus(jobDepHistories);
                for (JobDepHistory jobDepHistory : jobDepHistories) {
                    List<JobRunHistory> jobRunHistoryList = jobRunHistoryMapper.selectList(new QueryWrapper<JobRunHistory>()
                            .eq("dep_oozie_job_id",jobDepHistory.getOozieJobId())
                            .eq("state",LastRunState.RUNNING.getCode()));
                    if(CollectionUtil.isNotEmpty(jobRunHistoryList)){
                        jobRunHistoryMapper.batchUpdateKillStatus(jobRunHistoryList);
                        for (JobRunHistory jobRunHistory : jobRunHistoryList) {
                            List<JobNodeRunHistory> jobNodeRunHistories = jobNodeRunHistoryMapper.selectList(new QueryWrapper<JobNodeRunHistory>()
                                    .eq("oozie_job_id", jobRunHistory.getOozieJobId())
                                    .eq("state",LastRunState.RUNNING.getCode()));
                            if(CollectionUtil.isNotEmpty(jobNodeRunHistories)){
                                for (JobNodeRunHistory jobNodeRunHistory : jobNodeRunHistories) {
                                    //改变其log中的运行状态
                                    if(null != jobNodeRunHistory){
                                        String logs = jobNodeRunHistory.getLog();
                                        if(StringUtils.isNotEmpty(logs)){
                                            try {
                                                JSONObject obj = JSONObject.parseObject(logs);
                                                if(null != obj){
                                                    String status = obj.getString("status");
                                                    if(StringUtils.isNotEmpty(status) && status.trim().toLowerCase().equals("running")){
                                                        obj.put("status","killed");
                                                        jobNodeRunHistory.setLog(obj.toJSONString());
                                                    }
                                                }
                                            }catch (Exception e){
                                                log.error("转换日志异常！error:{}",e.getMessage());
                                            }
                                        }
                                    }
                                }
                                jobNodeRunHistoryMapper.batchUpdateKillStatus(jobNodeRunHistories);
                            }
                        }
                    }
                }
            }
        }
    }


}
